package cn.doitedu.demo5;

import cn.doitedu.beans.EventBean;
import cn.doitedu.utils.StateValueCompareUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.longlong.Roaring64Bitmap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/14
 * @Desc: 学大数据，上多易教育
 * <p>
静态画像条件（画像库中有现成结果的）：
职业(tag_01_01) MATCH 宝妈
年龄(tag_01_02) >     35
近半年平均月消费额(tag_03_01) BETWEEN  1000-1200
动态画像统计条件：
A行为 3+ 次
[B行为，E行为 ，Q行为] 序列 2+ 次
规则触发条件：
X行为

{
"static_condition": [
{
"tag_name":"tag_01_01",
"compare_type":"MATCH",
"compare_value":["宝妈"]
},
{
"tag_name":"tag_01_02",
"compare_type":">",
"compare_value":[35]
},
{
"tag_name":"tag_03_01",
"compare_type":"BETWEEN",
"compare_value":[1000,2000]
}
],
"dynamic_cnt_condition": [
{
"event_id":"A",
"compare_type":">",
"compare_value":3
},
{
"event_id":"C",
"compare_type":">",
"compare_value":3
}
],
"dynamic_seq_condition":{
"event_seq":["B","E","Q"],
"compare_type":">=",
"compare_value":1
},
"trigger_event_id":"M"
}
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * -----------------------------------
 * <p>
 * 静态画像条件（画像库中有现成结果的）：
 * 职业（tag_01_01）：程序员
 * 年龄(tag_01_02)：22-28
 * 兴趣词包含(tag_04_01): 移动硬盘
 * <p>
 * 动态画像统计条件：
 * F行为 2+ 次
 * [B行为，D行为 ，K行为] 序列 1+ 次
 * <p>
 * 规则触发条件：
 * M行为
 * <p>
 * <p>
 * <p>
 * {
 * "static_condition": [
 * {
 * "tag_name":"tag_01_01",
 * "compare_type":"MATCH",
 * "compare_value":["程序员"]
 * },
 * {
 * "tag_name":"tag_01_02",
 * "compare_type":"BETWEEN",
 * "compare_value":[22,28]
 * },
 * {
 * "tag_name":"tag_04_01",
 * "compare_type":"MATCH",
 * "compare_value":["移动硬盘"]
 * }
 * ],
 * "dynamic_cnt_condition": [
 * {
 * "event_id":"F",
 * "compare_type":">=",
 * "compare_value":2
 * },
 * {
 * "event_id":"C",
 * "compare_type":">",
 * "compare_value":3
 * }
 * ],
 * <p>
 * "dynamic_seq_condition":{
 * "event_seq":["B","D","K"],
 * "compare_type":">=",
 * "compare_value":1
 * <p>
 * },
 * "trigger_event_id":"M"
 * }
 **/

@Slf4j
public class RuleModel_01_Calculator implements RuleCalculator {

    ValueState<Integer> idxState;
    ValueState<Integer> seqCntState;
    MapState<String, Integer> mapState;

    JSONObject message;

    String rule_id;

    JSONObject paramObject;

    Roaring64Bitmap preSelectUsersBitmap;

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson , Roaring64Bitmap preSelectUsersBitmap) {

        // 将预圈选人群bitmap，放到成员变量上
        this.preSelectUsersBitmap = preSelectUsersBitmap;


        // 解析规则参数
        paramObject = JSON.parseObject(ruleParamJson);

        rule_id = paramObject.getString("rule_id");

        mapState = runtimeContext.getMapState(new MapStateDescriptor<String, Integer>(rule_id + "-dy_cnt", String.class, Integer.class));

        idxState = runtimeContext.getState(new ValueStateDescriptor<Integer>(rule_id + "-idx", Integer.class));
        seqCntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(rule_id + "-seq", Integer.class));

        message = new JSONObject();
        message.put("rule_id", rule_id);

    }

    @Override
    public void calc(EventBean eventBean, Collector<String> out) throws Exception {

        long userId = eventBean.getUser_id();

        // 判断该行为的用户是否是本规则的预圈选人群，如果否，则直接返回，啥也不做
        if(!preSelectUsersBitmap.contains(userId)) return;


        String eventId = eventBean.getEvent_id();

        /**
         * 二、 对规则的动态画像 事件次数统计条件进行处理
         */
        JSONArray dcArray = paramObject.getJSONArray("dynamic_cnt_condition");
        for (int i = 0; i < dcArray.size(); i++) {
            JSONObject cntObject = dcArray.getJSONObject(i);
            String eid = cntObject.getString("event_id");

            if (eid.equals(eventId)) {
                // 从状态中取出当前条件事件的发生次数
                Integer oldValue = mapState.get(eid);
                oldValue = oldValue == null ? 0 : oldValue;

                // 更新本条件事件的发生次数
                mapState.put(eid, oldValue + 1);

            }

        }

        /**
         * 三、 对规则的动态画像 序列次数统计条件进行处理
         */
        JSONObject seqObject = paramObject.getJSONObject("dynamic_seq_condition");
        // 取出事件序列： [B ,D ,K]
        JSONArray eventSeq = seqObject.getJSONArray("event_seq");

        // 取出索引状态的值
        int idx = idxState.value() == null ? 0 : idxState.value();

        if (eventId.equals(eventSeq.getString(idx))) {
            idx++;
            if (idx == eventSeq.size()) {
                idx = 0;
                seqCntState.update(seqCntState.value() == null ? 1 : seqCntState.value() + 1);
            }
            idxState.update(idx);
        }


        /**
         * 四、 处理触发条件事件
         */
        String triggerEventId = paramObject.getString("trigger_event_id");


        if (triggerEventId.equals(eventId)) {
            // 判断 动态画像 事件次数 统计条件是否都满足
            for (int i = 0; i < dcArray.size(); i++) {

                JSONObject cntObject = dcArray.getJSONObject(i);
                String eid = cntObject.getString("event_id");
                String compareType = cntObject.getString("compare_type");
                double compareValue = cntObject.getDouble("compare_value");

                Integer stateValue = mapState.get(eid);

                boolean compare = StateValueCompareUtil.compare(stateValue, compareValue, compareType);

                if (!compare) return;
            }


            // 判断 动态画像  行为序列 次数 统计条件是否满足
            Integer stateValue = seqCntState.value();
            String compareType = seqObject.getString("compare_type");
            Double compareValue = seqObject.getDouble("compare_value");

            boolean compare = StateValueCompareUtil.compare(stateValue, compareValue, compareType);

            // 如果本条件还满足，则要触发消息
            if(compare){

                message.put("user_id",eventBean.getUser_id());
                message.put("hit_time",eventBean.getAction_time());

                out.collect(message.toJSONString());
            }

        }


    }



    @Override
    public void offline() {

        log.warn("规则运算机:{} ,被下线，准备清理状态",this.rule_id);

        idxState.clear();
        seqCntState.clear();
        mapState.clear();

    }

    @Override
    public String getRuleId() {
        return this.rule_id;
    }
}
